package com.atguigu.hbase;

import com.atguigu.util.HBaseUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.CollectionUtils;
import org.junit.Test;

import javax.imageio.ImageIO;
import java.awt.*;
import java.awt.image.BufferedImage;
import java.io.*;
import java.net.HttpURLConnection;
import java.net.ProtocolException;
import java.net.URL;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.List;

/**
 * @author lvbingbing
 * @date 2020-07-20 23:59
 */
@Slf4j
public class HBaseApi {

    /**
     * 创建表
     *
     * @throws IOException
     */
    @Test
    public void createTable() {
        // 获取配置
        Configuration configuration = HBaseConfiguration.create();

        // HBase配置信息可以通过两种方式获取：
        // 1、Java客户端从类路径里的 hbase-site.xml文件里获取配置信息
        // 2、在连接中显式设定配置信息来获取

        // zookeeper节点
        configuration.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");

        // zookeeper端口
        configuration.set("hbase.zookeeper.property.clientPort", "2181");

        try (
                // 创建连接
                Connection connection = ConnectionFactory.createConnection(configuration);
                // 管理员可用于创建，删除，列出，启用和禁用表，添加和删除表列族以及其他管理操作。
                Admin admin = connection.getAdmin()
        ) {
            TableName student = TableName.valueOf("student");
            if (!admin.tableExists(student)) {
                HTableDescriptor hTableDescriptor = new HTableDescriptor(student);
                HColumnDescriptor columnFamily1 = new HColumnDescriptor("info");
                // HBase定义表时只需要声明列族即可。往HBase写入数据时，字段可以动态、按需指定。
                hTableDescriptor.addFamily(columnFamily1);
                HColumnDescriptor columnFamily2 = new HColumnDescriptor("account");
                hTableDescriptor.addFamily(columnFamily2);
                admin.createTable(hTableDescriptor);
                // 验证表是否创建成功
                HBaseUtil.validateTable(connection, admin, student);
            } else {
                log.warn("{}表已经存在，无需重复创建！", student.getNameAsString());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 删除表
     */
    @Test
    public void dropTable() {
        Configuration configuration = HBaseConfiguration.create();

        configuration.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
        configuration.set("hbase.zookeeper.property.clientPort", "2181");

        try (
                Connection connection = ConnectionFactory.createConnection(configuration);
                Admin admin = connection.getAdmin()
        ) {
            TableName[] tableNames = admin.listTableNames();
            for (TableName name : tableNames) {
                log.info("表名称：{}", name);
            }
            TableName student = TableName.valueOf("student");
            if (admin.tableExists(student)) {
                admin.disableTable(student);
                admin.deleteTable(student);
                log.info("{}表删除成功！", student.getNameAsString());
            } else {
                log.warn("{}表不存在，无法删除！", student.getNameAsString());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 插入数据。HBase插入数据叫put，不是insert
     */
    @Test
    public void addCell() {
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
        configuration.set("hbase.zookeeper.property.clientPort", "2181");

        try (
                Connection connection = ConnectionFactory.createConnection(configuration);
                Table student = connection.getTable(TableName.valueOf("student"))
        ) {
            addStudent(student, "1001", "info", "name", "张三");
            addStudent(student, "1001", "info", "age", "28");
            addStudent(student, "1001", "account", "accountId", "62271932004389549");
            addStudent(student, "1002", "info", "name", "李四");
            addStudent(student, "1002", "info", "age", "36");
            addStudent(student, "1002", "account", "accountId", "62232930843948509");
            validateAdd(student, "1001", "1002");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 删除数据
     */
    @Test
    public void deleteCell() {

        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
        configuration.set("hbase.zookeeper.property.clientPort", "2181");

        try (
                Connection connection = ConnectionFactory.createConnection(configuration);
                Table student = connection.getTable(TableName.valueOf("student"))
        ) {
            Delete delete = new Delete("1001".getBytes());
            delete.addColumn("info".getBytes(), "name".getBytes());
            delete.addColumn("account".getBytes(), "accountId".getBytes());
            student.delete(delete);

            validateDelete(student, "1001", "info", "name");
            validateDelete(student, "1001", "account", "accountId");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 删除多行数据
     */
    @Test
    public void deleteRows() {
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
        configuration.set("hbase.zookeeper.property.clientPort", "2181");

        try (
                Connection connection = ConnectionFactory.createConnection(configuration);
                Table student = connection.getTable(TableName.valueOf("student"))
        ) {
            List<Delete> deletes = new ArrayList<>();
            Delete delete1 = new Delete("1001".getBytes());
            Delete delete2 = new Delete("1002".getBytes());
            deletes.add(delete1);
            deletes.add(delete2);
            student.delete(deletes);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 查数据一：get
     */
    @Test
    public void getCell() {

        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
        configuration.set("hbase.zookeeper.property.clientPort", "2181");

        try (
                Connection connection = ConnectionFactory.createConnection(configuration);
                Table student = connection.getTable(TableName.valueOf("student"));
        ) {
            Get get = new Get("1001".getBytes());
            get.addFamily("info".getBytes());
            Result result = student.get(get);
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                String columnFamily = Bytes.toString(CellUtil.cloneFamily(cell));
                String columnQualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                log.info("rowKey：{}, columnFamily：{}, columnQualifier：{}, columnValue：{}", rowKey, columnFamily, columnQualifier, columnValue);
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 查数据二：scan
     */
    @Test
    public void scanCell() {
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
        configuration.set("hbase.zookeeper.property.clientPort", "2181");
        try (
                Connection connection = ConnectionFactory.createConnection(configuration);
                Table student = connection.getTable(TableName.valueOf("student"));
        ) {
            Scan scan = new Scan("1001".getBytes(), "1003".getBytes());
            ResultScanner scanner = student.getScanner(scan);
            for (Result result : scanner) {
                Cell[] cells = result.rawCells();
                for (Cell cell : cells) {
                    String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                    String columnFamily = Bytes.toString(CellUtil.cloneFamily(cell));
                    String columnQualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                    log.info("rowKey：{}, columnFamily：{}, columnQualifier：{}, columnValue：{}", rowKey, columnFamily, columnQualifier, columnValue);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 验证数据是否删除成功
     *
     * @param student         表
     * @param columnFamily    列族
     * @param columnQualifier 列限定符
     */
    private void validateDelete(Table student, String rowKey, String columnFamily, String columnQualifier) throws IOException {
        Get get = new Get(rowKey.getBytes());
        get.addFamily(columnFamily.getBytes());
        get.addColumn(columnFamily.getBytes(), columnQualifier.getBytes());
        Result result = student.get(get);
        List<Cell> columnCells = result.getColumnCells(columnFamily.getBytes(), columnQualifier.getBytes());
        if (CollectionUtils.isEmpty(columnCells)) {
            log.info("删除成功，未查询到rowKey = {}, 列族 = {}, 列限定符 = {}的数据！", rowKey, columnFamily, columnQualifier);
        } else {
            log.error("删除失败！rowKey = {}, 列族 = {}, 列限定符 = {}", rowKey, columnFamily, columnQualifier);
        }
    }

    /**
     * 根据传入的表、列键、列族、列限定符、列值 新增一条数据
     *
     * @param student         表
     * @param rowKey          列键。一条数据的唯一标识
     * @param columnFamily    列族
     * @param columnQualifier 列限定符
     * @param columnValue     列值
     */
    private void addStudent(Table student, String rowKey, String columnFamily, String columnQualifier, String columnValue) throws IOException {

        // {rowKey, column Family：column Qualifier, time Stamp} 唯一确定的单元。
        // cell中的数据是没有类型的，全部是字节码形式存储。
        Cell cell = new KeyValue(rowKey.getBytes(), columnFamily.getBytes(), columnQualifier.getBytes(), columnValue.getBytes());

        Put put = new Put(rowKey.getBytes());
        put.add(cell);
        student.put(put);
    }

    /**
     * @param student 表
     * @param rowKeys 列键
     * @throws IOException
     */
    private void validateAdd(Table student, String... rowKeys) throws IOException {
        for (String rowKey : rowKeys) {
            Get get = new Get(rowKey.getBytes());
            Result result = student.get(get);
            if (!result.isEmpty()) {
                List<Cell> cells = result.listCells();
                for (Cell cell1 : cells) {
                    String rowKey1 = Bytes.toString(CellUtil.cloneRow(cell1));
                    String columnFamily1 = Bytes.toString(CellUtil.cloneFamily(cell1));
                    String columnQualifier1 = Bytes.toString(CellUtil.cloneQualifier(cell1));
                    String columnValue1 = Bytes.toString(CellUtil.cloneValue(cell1));
                    log.info("数据插入成功！刚新增的一条数据为：rowKey：{}，列族：{}，列限定符：{}，列值：{}", rowKey1, columnFamily1, columnQualifier1, columnValue1);
                }
            }
        }
    }
}
